package mapred;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.Date;
import java.util.Scanner;
import java.util.concurrent.Semaphore;

import mapred.config.AppConfig;
import mapred.config.CombineConfig;
import mapred.config.InputConfig;
import mapred.config.JobConfig;
import mapred.config.MapConfig;
import mapred.config.OutputConfig;
import mapred.config.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.Counters.Group;

public abstract class Job extends Thread {

    private static final String LOG_FILE = "log.log";

    private Semaphore sem;

    private String jobName;

    private AppConfig app = new AppConfig();

    public Job(String jobName, String... settings) throws IOException {
        this(jobName, new Semaphore(1), settings);
    }

    public Job(String jobName, Semaphore sem, String... settings)
            throws IOException {
        super();
        this.jobName = jobName.replaceAll("\\s+", "_");
        this.sem = sem;
        app.set("-timeout", 600000);
        app.set("-vmargs", "-Xmx2g");
        app.set("-overwrite", "no");
        app.set(settings);

        logCommand();
    }

    private void logCommand() throws IOException {
        BufferedWriter bw = new BufferedWriter(new FileWriter(LOG_FILE, true));
        bw.write("[SUBMIT] " + new Date() + "\t");
        bw.write(this.jobName + ": " + this.getClass().getSimpleName());
        for (String a : app) {
            bw.write(" " + a + "=" + app.get(a));
        }
        bw.newLine();
        bw.newLine();
        bw.flush();
        bw.close();
    }

    public void run() {
        try {
            sem.acquire();

            String start = new Date().toString();

            JobConfig config = new JobConfig();
            config.getJobConf().setJobName(jobName);
            config.set("mapred.task.timeout", app.get("-timeout"));
            config.set("mapred.child.java.opts", app.get("-vmargs"));
            FileSystem fs = config.getFileSystem();
            setUp(config, fs);

            JobConf conf = config.getJobConf();
            setOutput(conf, fs);
            setReduce(conf);
            setCombine(conf);
            setMap(conf);
            setInput(conf, fs);

            putAppToConf(conf);
            
            System.out.println(jobName);
            System.out.println(app);
        
            RunningJob job = JobClient.runJob(config.getJobConf());
            tearDown(config, fs, job);

            String logPathStr = this.getLogPath();
            if (logPathStr != null) {
                Path logPath = new Path(logPathStr);
                BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(
                        fs.create(new Path(logPath, job.getJobName() + "_"
                                + job.getID().toString()))));
                writeJobCountLog(job, start, bw);
                bw = new BufferedWriter(new FileWriter(LOG_FILE, true));
                writeJobCountLog(job, start, bw);
            }

            sem.release();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void putAppToConf(JobConf conf) {
        for (String k : app) {
            if (k.startsWith("-")) {
                conf.set(k.substring(1), app.get(k));
            }
        }
    }

    private void writeJobCountLog(RunningJob job, String start,
            BufferedWriter bw) throws IOException {
        bw.write(job.getJobName());
        bw.newLine();
        bw.write(job.getID().toString());
        bw.newLine();
        bw.write(start);
        bw.newLine();
        bw.write(new Date().toString());
        bw.newLine();
        bw.write(job.getCounters().toString());
        bw.newLine();
        bw.newLine();
        bw.flush();
        bw.close();
    }

    private void setOutput(JobConf conf, FileSystem fs) throws IOException {
        OutputConfig oc = this.getOutputConfig(fs);
        oc.replace(app);

        conf.setOutputFormat(oc.getOutputFormat());
        conf.setOutputKeyClass(oc.getOutputKeyClass());
        conf.setOutputValueClass(oc.getOutputValueClass());
        Path outputPath = new Path(oc.getOutputPath());
        FileOutputFormat.setOutputPath(conf, outputPath);
        System.out.println("Output Path: " + outputPath);

        if (fs.exists(outputPath)) {
            if (this.getUserSetting("-overwrite").toLowerCase().startsWith("y")) {
                fs.delete(outputPath, true);
            } else {
                System.out.println("Output Path " + outputPath.toString()
                        + " already exists, overwrite it?(y/n)");
                Scanner sc = new Scanner(System.in);
                if (sc.next().trim().toLowerCase().startsWith("y")) {
                    fs.delete(outputPath, true);
                } else {
                    System.exit(0);
                }
            }
        }

        if (oc.isCompressOutput()) {
            FileOutputFormat.setCompressOutput(conf, true);
            FileOutputFormat.setOutputCompressorClass(conf, oc
                    .getOutputCompressorClass());
        } else {
            FileOutputFormat.setCompressOutput(conf, false);
        }
    }

    private void setReduce(JobConf conf) {
        ReduceConfig rc = this.getReduceConfig();
        rc.replace(app);

        conf.setReducerClass(rc.getReducerClass());
        conf.setNumReduceTasks(rc.getNumReduceTasks());

        conf.setMaxReduceAttempts(10);
    }

    private void setCombine(JobConf conf) {
        CombineConfig cc = this.getCombineConfig();
        if (cc.getCombinerClass() != null) {
            conf.setCombinerClass(cc.getCombinerClass());
        }
    }

    private void setMap(JobConf conf) {
        MapConfig mc = this.getMapConfig();
        conf.setMapperClass(mc.getMapperClass());
        conf.setNumMapTasks(0);
        conf.setMapOutputKeyClass(mc.getOutputKeyClass());
        conf.setMapOutputValueClass(mc.getOutputValueClass());

        conf.setMaxMapTaskFailuresPercent(1);
        conf.setMaxMapAttempts(10);
    }

    private void setInput(JobConf conf, FileSystem fs) throws IOException {
        InputConfig ic = this.getInputConfig(fs);
        ic.replace(app);

        conf.setInputFormat(ic.getInputFormat());
        String[] paths = ic.getInputPaths();
        for (String path : paths) {
            Path inputpath = new Path(path);
            if (fs.exists(inputpath)) {
                FileInputFormat.addInputPath(conf, inputpath);
                System.out.println("Input Path: " + inputpath);
            } else {
                System.out.println("Cannot Find Path " + inputpath);
            }
        }
    }

    public void submit() throws Exception {
        this.start();
        Thread.sleep(1000);
        sem.acquire();
    }

    protected abstract void tearDown(JobConfig conf, FileSystem fs,
            RunningJob job) throws IOException;

    protected abstract void setUp(JobConfig conf, FileSystem fs)
            throws IOException;

    protected abstract OutputConfig getOutputConfig(FileSystem fs)
            throws IOException;

    protected abstract ReduceConfig getReduceConfig();

    protected abstract CombineConfig getCombineConfig();

    protected abstract MapConfig getMapConfig();

    protected abstract InputConfig getInputConfig(FileSystem fs)
            throws IOException;

    private String getLogPath() {
        return "/logs/" + this.jobName + "/";
    }

    public String getUserSetting(String key) {
        return app.get(key);
    }

    public void forceUserSetting(String key, Object value) {
        app.set(key, value);
    }

    public long getCounter(RunningJob job, String group, String name) {
        try {
            return job.getCounters().findCounter(group, name).getCounter();
        } catch (Exception e) {
            return 0;
        }
    }

    public long getCounterSumByGroup(RunningJob job, String group) {
        long ret = 0;
        try {
            Group g = job.getCounters().getGroup(group);
            for (Counter c : g) {
                ret += c.getCounter();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return ret;
    }
    
    public static void buildXML(Class<?> cls) throws Exception {
        AntBuilder.buildXML(cls);
        System.exit(0);
    }

}
